package com.shujia.flink.core;

import com.shujia.flink.event.MyEvent;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class Demo05WaterMarkStrategy {
    public static void main(String[] args) throws Exception {
        // 自定义水位线策略
        // 参考链接：https://blog.csdn.net/zznanyou/article/details/121666563
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        DataStreamSource<String> eventDS = env.socketTextStream("master", 8888);

        // 在Source之后就指定水位线策略
        eventDS.assignTimestampsAndWatermarks(new WatermarkStrategy<String>() {
                    // 指定时间戳的提取策略
                    @Override
                    public TimestampAssigner<String> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
                        return new SerializableTimestampAssigner<String>() {
                            @Override
                            public long extractTimestamp(String element, long recordTimestamp) {
                                return Long.parseLong(element.split(",")[1]);
                            }
                        };
                        // 简写方式
//                return (ele,ts)->Long.parseLong(ele.split(",")[1]);
                    }

                    // 指定水位线的策略
                    @Override
                    public WatermarkGenerator<String> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                        return new MyWatermarkGenerator();
                    }
                })
                // 将数据变成KV格式，即：单词,1
                .map(line -> Tuple2.of(line.split(",")[0], 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                // 当窗口满足执行条件：1、水位线超过了窗口的结束时间 2、窗口有数据 触发一次process方法
                .process(new ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow>.Context context, Iterable<Tuple2<String, Integer>> elements, Collector<Tuple2<String, Integer>> out) throws Exception {
                        System.out.println("窗口触发执行了。");
                        System.out.println("当前水位线为：" + context.currentWatermark() + ",当前窗口的开始时间：" + context.window().getStart() + ",当前窗口的结束时间：" + context.window().getEnd());

                        // 基于elements做统计 通过out可以将结果发送到下游
                    }
                }).print();

//        // 将每条数据变成MyEvent类型
//        eventDS.map(new MapFunction<String, MyEvent>() {
//            @Override
//            public MyEvent map(String value) throws Exception {
//                String[] split = value.split(",");
//                return new MyEvent(split[0],Long.parseLong(split[1]));
//            }
//        }).assignTimestampsAndWatermarks(new WatermarkStrategy<MyEvent>() {
//            @Override
//            public TimestampAssigner<MyEvent> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
//                return new SerializableTimestampAssigner<MyEvent>() {
//                    @Override
//                    public long extractTimestamp(MyEvent element, long recordTimestamp) {
//                        return element.getTs();
//                    }
//                };
//            }
//
//            @Override
//            public WatermarkGenerator<MyEvent> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
//                return new MyMapWatermarkGenerator();
//            }
//        }).keyBy(my->my.word)
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
//                // 当窗口满足执行条件：1、所有线程的水位线都超过了窗口的结束时间 2、窗口有数据 触发一次process方法
//                .process(new ProcessWindowFunction<MyEvent, String, String, TimeWindow>() {
//                    @Override
//                    public void process(String s, ProcessWindowFunction<MyEvent, String, String, TimeWindow>.Context context, Iterable<MyEvent> elements, Collector<String> out) throws Exception {
//                        System.out.println("窗口触发执行了。");
//                        System.out.println("当前水位线为：" + context.currentWatermark() + ",当前窗口的开始时间：" + context.window().getStart() + ",当前窗口的结束时间：" + context.window().getEnd());
//
//                        // 基于elements做统计 通过out可以将结果发送到下游
//                    }
//                }).print();



        env.execute();

    }
}

// 用于map之后指定水位线生成策略
class MyMapWatermarkGenerator implements WatermarkGenerator<MyEvent> {

    private final long maxOutOfOrderness = 0;

    private long currentMaxTimeStamp;

    // 每来一条数据会处理一次
    @Override
    public void onEvent(MyEvent event, long eventTimestamp, WatermarkOutput output) {
        currentMaxTimeStamp = Math.max(currentMaxTimeStamp, eventTimestamp);
        System.out.println("当前线程编号为：" + Thread.currentThread().getId() + ",当前水位线为：" + (currentMaxTimeStamp - maxOutOfOrderness));
    }

    // 周期性的执行：env.getConfig().getAutoWatermarkInterval(); 默认是200ms
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(currentMaxTimeStamp - maxOutOfOrderness));
    }
}

// 用于Source之后直接指定水位线生成策略
class MyWatermarkGenerator implements WatermarkGenerator<String> {

    private final long maxOutOfOrderness = 0;

    private long currentMaxTimeStamp;

    // 每来一条数据会处理一次
    @Override
    public void onEvent(String event, long eventTimestamp, WatermarkOutput output) {
        currentMaxTimeStamp = Math.max(currentMaxTimeStamp, eventTimestamp);
        System.out.println("当前线程编号为：" + Thread.currentThread().getId() + ",当前水位线为：" + (currentMaxTimeStamp - maxOutOfOrderness));
    }

    // 周期性的执行：env.getConfig().getAutoWatermarkInterval(); 默认是200ms
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(currentMaxTimeStamp - maxOutOfOrderness));
    }
}


